package com.shujia.flink.core

import org.apache.flink.api.scala._

object Demo2BacthWordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 创建flink batch 环境
      *
      */

    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment


    /**
      * 读取数据
      *
      */

    val lineDS: DataSet[String] = env.readTextFile("flink/data/words.txt")

    val words: DataSet[String] = lineDS.flatMap(_.split(","))

    val kvDS: DataSet[(String, Int)] = words.map((_, 1))

    //指定分组字段
    val groupBuDS: GroupedDataSet[(String, Int)] = kvDS.groupBy(0)

    val countDS: AggregateDataSet[(String, Int)] = groupBuDS.sum(1)


    val resultDS: DataSet[String] = countDS.map(kv => s"${kv._1},${kv._2}")


    resultDS.writeAsText("flink/data/out1")

    env.execute()

  }
}
